package 实训二;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyClickHouseUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArIsNewPageWindow2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> pageDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "xxxx"));
        DataStreamSource<String> uvDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_unique_visitor_detail", "xxxx"));
        DataStreamSource<String> ujDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_user_jump_detail", "xxxx"));

        SingleOutputStreamOperator<TrafficPageViewBean> uvmap = uvDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null, vc, ch, ar,
                        isNew, 1L, 0L, 0L, 0L, 0L, jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> ujmap = ujDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null, vc, ch, ar, isNew,
                        0L, 0L, 0L, 0L, 1L, jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> pagemap = pageDs.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                Long sv = 0L;
                if (last_page_id == null) {
                    sv = 1L;
                }
                return new TrafficPageViewBean(null, null, vc, ch, ar, isNew,
                        0L, sv, 1L, page.getLong("during_time"), 0L, jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> union = pagemap.union(uvmap, ujmap).assignTimestampsAndWatermarks(WatermarkStrategy.
                <TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                        return trafficPageViewBean.getTs();
                    }
                }));
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream =
                union.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean t1)
                            throws Exception {
                        return Tuple4.of(t1.getVc(), t1.getCh(), t1.getAr(), t1.getIsNew());
                    }
                });
        SingleOutputStreamOperator<TrafficPageViewBean> aggregate = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L))).allowedLateness(Time.seconds(10L))
                .aggregate(new AggregateFunction<TrafficPageViewBean, Tuple5<Long, Long, Long, Long, Long>, TrafficPageViewBean>() {

                    //初始化累加器
                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> createAccumulator() {

                        return Tuple5.of(0L, 0L, 0L, 0L, 0L);
                    }

                    // 每条数据都走这里
                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> add(TrafficPageViewBean t1,
                                                                    Tuple5<Long, Long, Long, Long, Long> LongTuple5) {
                        return Tuple5.of(t1.getUvCt() + LongTuple5.f0,
                                t1.getSvCt() + LongTuple5.f1,
                                t1.getPvCt() + LongTuple5.f2,
                                t1.getDurSum() + LongTuple5.f3,
                                t1.getUjCt() + LongTuple5.f4);
                    }

                    //最终结果输出
                    @Override
                    public TrafficPageViewBean getResult(Tuple5<Long, Long, Long, Long, Long> LongTuple5) {
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean();
                        trafficPageViewBean.setUvCt(LongTuple5.f0);
                        trafficPageViewBean.setSvCt(LongTuple5.f1);
                        trafficPageViewBean.setPvCt(LongTuple5.f2);
                        trafficPageViewBean.setDurSum(LongTuple5.f3);
                        trafficPageViewBean.setUjCt(LongTuple5.f4);
                        return trafficPageViewBean;
                    }

                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> merge(Tuple5<Long, Long, Long, Long, Long> longLongLongLongLongTuple5, Tuple5<Long, Long, Long, Long, Long> acc1) {
                        return null;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                                      TimeWindow window, Iterable<TrafficPageViewBean> input,
                                      Collector<TrafficPageViewBean> out) throws Exception {
                        //获取累加器值 只有一条
                        TrafficPageViewBean next1 = input.iterator().next();
                        //获取key Tuple4.of(t1.getVc(), t1.getCh(), t1.getAr(), t1.getIsNew());
                        String Vc = stringStringStringStringTuple4.f0;
                        String Ch = stringStringStringStringTuple4.f1;
                        String Ar = stringStringStringStringTuple4.f2;
                        String IsNew = stringStringStringStringTuple4.f3;
                        //获取窗口
                        Timestamp start = new Timestamp(window.getStart());
                        Timestamp end = new Timestamp(window.getEnd());
                        //ts 当前时间
                        //     trafficPageViewBean.setUvCt(LongTuple5.f0);
                        //                        trafficPageViewBean.setSvCt(LongTuple5.f1);
                        //                        trafficPageViewBean.setPvCt(LongTuple5.f2);
                        //                        trafficPageViewBean.setDurSum(LongTuple5.f3);
                        //                        trafficPageViewBean.setUjCt(LongTuple5.f4);
                        Long ts = System.currentTimeMillis();
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(start, end, Vc, Ch, Ar, IsNew, next1.getUvCt(), next1.getSvCt(), next1.getPvCt(), next1.getDurSum(), next1.getUjCt(), ts);
                        out.collect(trafficPageViewBean);
                    }
                });
        aggregate.print();
        aggregate.addSink(MyClickHouseUtil.getSinkFunction("insert into table gmall.dws_traffic_vc_ch_page_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));
        env.execute("DwsTrafficVcChArIsNewPageWindow2");
        env.execute();
    }
}
